package com.simoniu.flink.scala.tablesql

import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.bridge.scala.{BatchTableEnvironment, StreamTableEnvironment}

/**
 * 将DataSet转换成表
 * Created by simoniu
 */
object DataSetToTableScalaDemo {
  def main(args: Array[String]): Unit = {
    //获取BatchTableEnvironment
    val bbEnv = ExecutionEnvironment.getExecutionEnvironment
    val bbTableEnv = BatchTableEnvironment.create(bbEnv)

    //获取DataSet
    import org.apache.flink.api.scala._
    val set = bbEnv.fromCollection(Array((1, "jack", "female"), (2, "tom", "male"), (3, "mike", "male")))

    //第一种：将DataSet转换为view视图
    import org.apache.flink.table.api._
    bbTableEnv.createTemporaryView("myTable", set, 'id, 'name, 'gender)
    bbTableEnv.sqlQuery("select * from myTable where id > 1").execute().print()

    //第二种：将DataSet转换为table对象
    val table = bbTableEnv.fromDataSet(set, $"id", $"name", $"gender")
    table.select($"id", $"name", $"gender")
      .filter($"id" > 1)
      .execute()
      .print()

    //注意：'id,'name 和 $"id", $"name" 这两种写法是一样的效果
  }
}
